/************************************************************************
Copyright 2003, 2004, 2005, 2006, 2007, 2008 Grid Technology Research Center,
National Institute of Advanced Industrial Science and Technology.

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
************************************************************************/

package jp.aist.gtrc.plus.reserve;

import java.util.Calendar;
import java.util.Collection;
import java.util.LinkedList;

import jp.aist.gtrc.plus.reserve.ReserveStatus.State;
import static jp.aist.gtrc.plus.reserve.ReserveStatus.State.*;
import jp.aist.gtrc.plus.scheduler.node.NodeAllocateRequest;
import jp.aist.gtrc.plus.scheduler.node.NodeAllocator;
import jp.aist.gtrc.plus.scheduler.node.NodeInfo;
import jp.aist.gtrc.plus.scheduler.node.NodeManager;
import jp.aist.gtrc.plus.scheduler.util.CollectionUtil;

class ReserveNodeNormalizer {

	private final ReserveManager rsvMgr;
	private final ReserveTable table;
	private final ReserveRecorder recorder;
	private final NodeManager nodeMgr;
	private final NodeAllocator nodeAllocator;

	ReserveNodeNormalizer(ReserveManager rsvMgr) {

		this.rsvMgr = rsvMgr;
		this.table = rsvMgr.getReserveTable();
		this.recorder = rsvMgr.getReserveRecorder();
		this.nodeMgr = rsvMgr.getNodeManager();
		this.nodeAllocator = rsvMgr.getNodeAllocator();

	}

	private Collection<NodeInfo> dropAllocatedNodes(Collection<NodeInfo> allNodes,
			Collection<NodeInfo> allocatedNodes) {

		if (CollectionUtil.isEmpty(allocatedNodes)) {
			return allNodes;
		}

		LinkedList<NodeInfo> list = new LinkedList<NodeInfo>();
		for (NodeInfo node : allNodes) {
			// NOTE: allocatedNodes was got from ReserveNodeSer#getAll(),
			// which is LinkedHashSet object.
			// So we can check whether contains() or not at O(1).
			if (allocatedNodes.contains(node) == false) {
				list.add(node);
			}
		}

		return list;

	}

	protected ReserveNodeSet makeReserveNodeSet(ReserveRequest rsvReq) throws ReserveException {

		Calendar start = rsvReq.getStartTime();
		ReserveNodeSet nodeSet = new ReserveNodeSet(nodeMgr);
		Collection<NodeInfo> candidates, usable;
		NodeAllocateRequest allocReq = new NodeAllocateRequest(rsvReq);

		/*
		 * 1st: allocate nodes from previous reserves
		 */
		for (ReserveInfo info : table.getNotAfterEndTime(start)) {
			State state = info.getReserveState();
			if ((state != Confirmed) && (state != Running)) {
				continue;	// 'state' is done/canceled/error
			}

			candidates = dropAllocatedNodes(info.getReservedNodeInfos(), nodeSet.getAll());
			if (candidates.size() > 0) {
				usable = nodeAllocator.suitable(candidates, allocReq);
				if (usable != null) {
					nodeSet.add(usable);
				}
			}
			if (allocReq.isAllocatedEnough()) {
				return nodeSet;	// already allocated enough
			}

			ReserveNodeSet set = info.getReserveNodeSet();
			assert(set != null);
			if (set.isNodeSkipped() == false) {
				break;
			}
		}

		/*
		 * 2nd: allocate nodes from all nodes if 1st is not enough
		 */
		candidates = dropAllocatedNodes(nodeMgr.getNodes(), nodeSet.getAll());
		usable = null;
		if (candidates.size() > 0) {
			usable = nodeAllocator.suitable(candidates, allocReq);
		}
		if (allocReq.isAllocatedEnough()) {
			nodeSet.add(usable);
		} else {
			// we cannot allocate enough nodes
			throw new ReserveException("Cannot reserve nodes");
		}

		return nodeSet;

	}

	protected void reallocateNodes(Calendar startTime, boolean equalTimeAlso) throws ReserveException {

		Collection<ReserveInfo> candidateInfos;

		candidateInfos = table.getAfterStartTime(startTime, equalTimeAlso);
		if (CollectionUtil.isEmpty(candidateInfos)) {
			// no reserve info to reallocate
			return;
		}

		Collection<ReserveInfo> targetInfos = new LinkedList<ReserveInfo>();
		int nSkip = 0;

		/*
		 * 1st: free all reserved nodes which starts after 'rsvInfo'
		 */
		for (ReserveInfo info : candidateInfos) {
			if (info.getReserveState() != Confirmed) {
				continue;
			}

			ReserveNodeSet curNodeSet = info.getReserveNodeSet();
			assert(curNodeSet != null);
			if (curNodeSet.isNodeSkipped() == false) {
				nSkip++;
				if (nSkip >= 2) {
					break;
				}
			}

			targetInfos.add(info);
			info.unregisterReservedNodesTemporary();
		}

		if (targetInfos.size() == 0) {
			return;
		}
		rsvMgr.getCurrentOperation().setReallocatedReserveInfo(targetInfos);

		/*
		 * 2nd: re-allocate nodes for them
		 */
		for (ReserveInfo info : targetInfos) {
			ReserveRequest rsvReq = info.getReserveRequest();
			ReserveNodeSet newNodeSet = makeReserveNodeSet(rsvReq);
			if (info.modify(newNodeSet) == true) {
				recorder.store(info);
			}
		}

		/*
		 * NOTE: we don't recover modification here even if failed.
		 * Caller must use abort() to recover.
		 */
	}

	protected void reallocateJustBeforeStart(ReserveInfo rsvInfo) {

		/*
		 * Check whether all reserved nodes are alive now.
		 */
		ReserveRequest rsvReq = rsvInfo.getReserveRequest();
		ReserveNodeSet rsvNodeSet = rsvInfo.getReserveNodeSet();
		ReserveNodeAllocator rsvNodeAllocator = (ReserveNodeAllocator)nodeAllocator;
		NodeAllocateRequest allocReq = new NodeAllocateRequest(rsvReq);
		ReserveClassAdChecker checker = new ReserveClassAdChecker(
				nodeMgr, rsvNodeAllocator.getAdFilePath(),
				ReserveClassAdChecker.DEFAULT_EVAL_ATTR_RUNNING, allocReq);

		LinkedList<NodeInfo> okNodes = new LinkedList<NodeInfo>();
		LinkedList<NodeInfo> downNodes = new LinkedList<NodeInfo>();
		for (NodeInfo node : rsvNodeSet.getAll()) {
			boolean isOK = checker.isOK(node, okNodes);
			if (isOK) {
				okNodes.add(node);
			} else {
				downNodes.add(node);
			}
		}
		if (downNodes.size() == 0) {
			// All pre-reserved nodes are usable now.
			// Node reallocation isn't necessary.
			return;
		}

		/*
		 * Make a reservation request only for single node
		 */
		ReserveRequest singleReq;
		try {
			singleReq = new ReserveRequest(
					rsvReq.getOwner(), rsvReq.getUsers(), 1,
					rsvReq.getStartTime(), rsvReq.getEndTime());
		} catch (ReserveException e) {
			assert(false);
			return;
		}
		NodeAllocateRequest singleAllocReq = new NodeAllocateRequest(singleReq);

		/*
		 * Find idle unreserved nodes for down nodes one by one
		 */
		Collection<NodeInfo> candidates, usable;
		candidates = dropAllocatedNodes(nodeMgr.getNodes(), downNodes);
		if (candidates.size() == 0) {
			// cannot get idle node
			return;
		}
		for (NodeInfo downNode : downNodes) {
			usable = nodeAllocator.suitable(candidates, singleAllocReq);
			if (CollectionUtil.isEmpty(usable) == false) {
				// swap PhysicalNode assigned for downNode and idleNode
				NodeInfo idleNode = usable.iterator().next();
				assert(idleNode != null);
				nodeMgr.changeNodeInfoPair(downNode, idleNode);

				// remove 'idleNode' from candidates list
				// because it's reallocated now.
				candidates.remove(idleNode);
			}
		}

		// re-set reserved node names.
		// ReserveNodeSet and ReserveStatus (parent of ReserveInfo) have
		// node names as simple String array for db4o.
		rsvInfo.resetReservedNodeNames();

	}

}
